package com.zjl.Kafka.第03章_操作kafka;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import org.junit.Test;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

/**
 * *************************************************************
 * kafka文件存储
 *      主题名-分区
 *              leader-epoch-checkpoint
 *              partition.metadata
 *                      存储了分区和主题名
 *              00000000000000000000.log
 *                      默认 每1G 一个序号
 *                      log.segment.bytes
 *              00000000000000000000.index
 *                      偏移量索引文件  也叫稀疏索引文件
 *                      里面存了很多数据的 偏移量和数据在文件的位置
 *              00000000000000000000.timeindex
 *                      其他的只保存了   时间戳和偏移量
 * *************************************************************
 * 在 kafka 的 bin 目录
 * sh kafka-run-class.sh 执行的kafka内部的方法 入参 --print-data-log
 * sh kafka-run-class.sh kafka.tools.DumpLogSegments --files 主题文件路径 --print-data-log
 *      可以查看文件的内容
 *
 * *************************************************************
 *
 * 日志分段条件  00000000000000000000.log 添加新的序号   数字就是起始偏移量
 *      当前日志分段文件的大小超过了broker端参数 log.segment.bytes 配置的值。 log.segment.bytes 参数的默认值为1GB.
 *      当前日志分段中消息的最大时间戳与当前系统的时间戳的差值大于 log.roll.ms或log.roll.hours参数配置的值。
 *                  如果同时配置了log.roll.ms和log.roll.hours参数，那么log.roll.ms的优先级高。
 *                  默认情况下，只配置了log.roll.hours 参数,其值为168,即7天。
 *      偏移量索引文件或时间戳索引文件的大小达到 broker 端参数log.index.size.max.bytes配置的值。
 *                  log.index.size.max.bytes 的默认值为10MB。
 *      追加的消息的偏移量与当前日志分段的偏移量之间的差值大于Integer.MAX_VALUE,
 *                  即要追加的消息的偏移量不能转变为相对偏移量（offset - baseOffset > Integer.MAX_VALUE)。
 * ****************************************************************
 * 找文件数据时，是
 *      先根据 00000000000000000000.timeindex 的时间戳和偏移量找到
 *       00000000000000000000.index  索引偏移文件（稀疏索引文件）跳跃表 的原理
 *      找到指定数据文件(00000000000000000000.log)的指定位置
 *
 *
 *
 */
public class D1_数据存储文件解析 {
    @Test
    public  void 生产者() throws ExecutionException, InterruptedException {
        Map<String,Object> config = new HashMap<>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"82.157.71.243:9092");//连接地址
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());//key 的 序列化
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());//value 的 序列化
        config.put(ProducerConfig.BATCH_SIZE_CONFIG, 2);//每一批大小  单位kb


        //创建生产者对象
        KafkaProducer<String,String> kafkaProducer = new KafkaProducer<>(config);
        try{
            for (int i = 0; i < 10; i++) {
                //创建数据
                ProducerRecord<String,String> record = new ProducerRecord<>(
                        "x03"//主题名称  发送不成功改动主题名，就行了
                        ,5//分区的值必须在创建的时候就存在，否则  会超时
                        ,"key" + i//数据的key
                        ,"value" + i//数据的value
                );
                //将数据发送到kafka   异步发送，其他线程发送
                Future<RecordMetadata> send = kafkaProducer.send(record, new Callback() {//回调方法
                    @Override
                    public void onCompletion(RecordMetadata metadata, Exception exception) {
                        System.out.println("发送成功！！主题:" + metadata.topic() + "\t分区：" + metadata.partition() + "\t偏移量：" + metadata.offset());
                    }
                });

                RecordMetadata recordMetadata = send.get();//获取数据，只有获取数据后，才会继续执行

                System.out.println(recordMetadata);

                System.out.println(record);
            }
            System.out.println(kafkaProducer);

        }catch (Exception e){
            e.printStackTrace();
        }finally {
            try { TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { throw new RuntimeException(e); }
            //关闭对象
            kafkaProducer.close();
        }



    }
}
